/*
 * Copyright (C) 2012 The Android Open Source Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
import { DISCARD_STREAM_TIMEOUT_MILLIS, ExchangeCodec } from "../http/ExchangeCodec"
import { RealConnection } from "../connection/RealConnection"
import { OkHttpClient } from "../OkHttpClient"
import { BufferedSink } from "../okio/BufferedSink"
import { BufferedSource } from "../okio/BufferedSource"
import { Request } from '../Request'
import { Headers } from "../Headers"
import { RequestLine } from "../http/RequestLine"
import { checkOffsetAndCount, EMPTY_HEADERS, headersContentLength, kotlinCheck, kotlinRequire } from "../utils/Utils"
import { StatusLine } from "../http/StatusLine"
import { HeadersReader } from "./HeadersReader"
import { Response, ResponseBuilder } from "../Response"
import { promisesBody, receiveHeaders } from "../http/HttpHeaders"
import { Source } from "../okio/Source"
import { HttpUrl } from "../HttpUrl"
import { Sink } from "../okio/Sink"
import { IllegalStateException, IOException, ProtocolException } from "../Error/error"
import { minOf } from "../utils/NumberUtils"

/**
 * A socket connection that can be used to send HTTP/1.1 messages. This class strictly enforces the
 * following lifecycle:
 *
 *  1. [Send request headers][writeRequest].
 *  2. Open a sink to write the request body. Either [known][newKnownLengthSink] or
 *     [chunked][newChunkedSink].
 *  3. Write to and then close that sink.
 *  4. [Read response headers][readResponseHeaders].
 *  5. Open a source to read the response body. Either [fixed-length][newFixedLengthSource],
 *     [chunked][newChunkedSource] or [unknown][newUnknownLengthSource].
 *  6. Read from and close that source.
 *
 * Exchanges that do not have a request body may skip creating and closing the request body.
 * Exchanges that do not have a response body can call
 * [newFixedLengthSource(0)][newFixedLengthSource] and may skip reading and closing that source.
 */
export class Http1ExchangeCodec implements  ExchangeCodec {

  static NO_CHUNK_YET = -1
  static STATE_IDLE = 0 // Idle connections are ready to write request headers.
  static STATE_OPEN_REQUEST_BODY = 1
  static STATE_WRITING_REQUEST_BODY = 2
  static STATE_READ_RESPONSE_HEADERS = 3
  static STATE_OPEN_RESPONSE_BODY = 4
  static STATE_READING_RESPONSE_BODY = 5
  static STATE_CLOSED = 6

  /** The client that configures this stream. May be null for HTTPS proxy tunnels. */
  client?: OkHttpClient
  /** The connection that carries this stream. */
  connection: RealConnection
  source: BufferedSource
  sink: BufferedSink

  constructor(client: OkHttpClient, connection: RealConnection, source: BufferedSource, sink: BufferedSink) {
    this.client = client
    this.connection = connection
    this.source = source
    this.sink = sink
    this.headersReader = new HeadersReader(this.source)
  }

  state = Http1ExchangeCodec.STATE_IDLE
  headersReader: HeadersReader

  ResponseIsChunked(response: Response):boolean{
    return response.header("Transfer-Encoding")?.toLowerCase() === "chunked"
  }

  RequestIsChunked(request: Request):boolean{
    return request.header("Transfer-Encoding")?.toLowerCase() === "chunked"
  }

  /**
   * Received trailers. Null unless the response body uses chunked transfer-encoding and includes
   * trailers. Undefined until the end of the response body.
   */
  _trailers: Headers | null = null

  /** Returns true if this connection is closed. */
  _isClosed: boolean
  get isClosed(){
    this._isClosed = this.state === Http1ExchangeCodec.STATE_CLOSED
    return this._isClosed
  }

    createRequestBody(request: Request, contentLength: number): Sink {
      if (request.body !== null && request.body !== undefined && request.body.isDuplex()){
        throw new ProtocolException("Duplex connections are not supported for HTTP/1")
      }else if (this.RequestIsChunked(request)){
        return this.newChunkedSink()
      }else if (contentLength !== -1){
        return this.newKnownLengthSink()
      }else{
        throw new IllegalStateException("Cannot stream a request body without chunked encoding or a known content length!")
      }
    }

    cancel() {
      this.connection.cancel()
    }

    /**
     * Prepares the HTTP headers and sends them to the server.
     *
     * For streaming requests with a body, headers must be prepared **before** the output stream has
     * been written to. Otherwise the body would need to be buffered!
     *
     * For non-streaming requests with a body, headers must be prepared **after** the output stream
     * has been written to and closed. This ensures that the `Content-Length` header field receives
     * the proper value.
     */
    writeRequestHeaders(request: Request) {
      let requestLine = RequestLine.get(request, this.connection.route().proxy.proxyType)
      this.writeRequest(request.headers, requestLine)
    }

    reportedContentLength(response: Response): number {
      if (!promisesBody(response)){
        return 0
      }else if (this.ResponseIsChunked(response)){
        return -1
      }else {
        return headersContentLength(response)
      }
    }

    openResponseBodySource(response: Response): Source {
      if (!promisesBody(response)){
        return this.newFixedLengthSource(0)
      }else if (this.ResponseIsChunked(response)){
        return this.newChunkedSource(response.request.url)
      }else{
        let contentLength = headersContentLength(response)
        if (contentLength !== -1){
          return this.newFixedLengthSource(contentLength)
        }else{
          return this.newUnknownLengthSource()
        }
      }
    }


    trailers(): Headers {
      kotlinCheck(this.state === Http1ExchangeCodec.STATE_CLOSED, "too early; can't read the trailers yet")
      if (this._trailers !== null && this._trailers !== undefined){
        return this._trailers
      }
      return EMPTY_HEADERS
    }

    flushRequest() {
      this.sink.flush()
    }

    async finishRequest() {
      await this.sink.flush()
    }

    /** Returns bytes of a request header for sending on an HTTP transport. */
    writeRequest(headers: Headers, requestLine: string) {
      kotlinCheck(this.state === Http1ExchangeCodec.STATE_IDLE, `state: ${this.state}`)
      this.sink.write(requestLine)
      this.sink.write("\r\n")
      for (let i = 0; i < headers.size; i++){
        this.sink.write(headers.name(i))
        this.sink.write(": ")
        this.sink.write(headers.value(i))
        this.sink.write("\r\n")
      }
      this.sink.write("\r\n")
      this.state = Http1ExchangeCodec.STATE_OPEN_REQUEST_BODY
    }

    async readResponseHeaders(expectContinue: boolean): Promise<ResponseBuilder | null>{
      kotlinCheck(this.state === Http1ExchangeCodec.STATE_OPEN_REQUEST_BODY || this.state === Http1ExchangeCodec.STATE_READ_RESPONSE_HEADERS, `state: ${this.state}`)
      try {
        let firstLine = await this.headersReader.readLine()
        let statusLine = StatusLine.parse(firstLine)

        let responseBuilder = new Response.Builder()
          .protocol(statusLine.protocol)
          .code(statusLine.code)
          .message(statusLine.message)
          .headers(await this.headersReader.readHeaders())
        if (expectContinue && statusLine.code == StatusLine.HTTP_CONTINUE){
          return null
        }else if (statusLine.code == StatusLine.HTTP_CONTINUE){
          this.state = Http1ExchangeCodec.STATE_READ_RESPONSE_HEADERS
          return responseBuilder
        }else{
          this.state = Http1ExchangeCodec.STATE_OPEN_RESPONSE_BODY
          return responseBuilder
        }
      } catch (e) {
        // Provide more context if the server ends the stream before sending a response.
        let address = this.connection.route().address.url.redact()
        throw new IOException(`unexpected end of stream on ${address}` + JSON.stringify(e))
      }
    }

    newChunkedSink(): Sink {
      kotlinCheck(this.state === Http1ExchangeCodec.STATE_OPEN_REQUEST_BODY,`state: ${this.state}`)
      this.state = Http1ExchangeCodec.STATE_WRITING_REQUEST_BODY
      return new ChunkedSink(this.sink, this)
    }

    newKnownLengthSink(): Sink {
      kotlinCheck(this.state === Http1ExchangeCodec.STATE_OPEN_REQUEST_BODY,`state: ${this.state}`)
      this.state = Http1ExchangeCodec.STATE_WRITING_REQUEST_BODY
      return new KnownLengthSink(this.sink, this)
    }

    newFixedLengthSource(length: number): Source {
      kotlinCheck(this.state === Http1ExchangeCodec.STATE_OPEN_RESPONSE_BODY,`state: ${this.state}`)
      this.state = Http1ExchangeCodec.STATE_READING_RESPONSE_BODY
      return new FixedLengthSource(this, length)
    }

    newChunkedSource(url: HttpUrl): Source {
      kotlinCheck(this.state === Http1ExchangeCodec.STATE_OPEN_RESPONSE_BODY,`state: ${this.state}`)
      this.state = Http1ExchangeCodec.STATE_READING_RESPONSE_BODY
      return new ChunkedSource(this, url)
    }

    private newUnknownLengthSource(): Source {
      kotlinCheck(this.state === Http1ExchangeCodec.STATE_OPEN_RESPONSE_BODY,`state: ${this.state}`)
      this.state = Http1ExchangeCodec.STATE_READING_RESPONSE_BODY
      this.connection.noNewExchanges()
      return new UnknownLengthSource(this)
    }

    /**
     * Sets the delegate of `timeout` to [Timeout.NONE] and resets its underlying timeout
     * to the default configuration. Use this to avoid unexpected sharing of timeouts between pooled
     * connections.
     */
    // detachTimeout(timeout: ForwardingTimeout) {
      // val oldDelegate = timeout.delegate
      // timeout.setDelegate(Timeout.NONE)
      // oldDelegate.clearDeadline()
      // oldDelegate.clearTimeout()
    // }

    /**
     * The response body from a CONNECT should be empty, but if it is not then we should consume it
     * before proceeding.
     */
    skipConnectBody(response: Response) {
      let contentLength = headersContentLength(response)
      if (contentLength == -1) return
      let body = this.newFixedLengthSource(contentLength)
      body.skipAll() // Int.MAX_VALUE, MILLISECONDS)
      body.close()
    }
}


/** An HTTP request body. */
export class KnownLengthSink implements Sink {
  // private timeout = ForwardingTimeout(sink.timeout())
  private closed: boolean = false
  // override timeout(): Timeout = timeout

  constructor(private sink: Sink, private codec: Http1ExchangeCodec) {
  }

  async write(data: Uint8Array | string, byteCount?: number) {
    kotlinCheck(!this.closed, "closed")
    checkOffsetAndCount(data.length, 0, byteCount)
    await this.sink.write(data, byteCount)
  }

  async flush() {
    if (this.closed) return // Don't throw; this stream might have been closed on the caller's behalf.
    await this.sink.flush()
  }

  async close() {
    if (this.closed) return
    this.closed = true
    // this.codec.detachTimeout(this.timeout)
    this.codec.state = Http1ExchangeCodec.STATE_READ_RESPONSE_HEADERS
  }

  buffer(){
    return this.sink
  }
}

/**
 * An HTTP body with alternating chunk sizes and chunk bodies. It is the caller's responsibility
 * to buffer chunks; typically by using a buffered sink with this sink.
 */
export class ChunkedSink implements Sink {
  // private val timeout = ForwardingTimeout(sink.timeout())
  private closed: boolean = false
  constructor(private sink: Sink, private codec: Http1ExchangeCodec) {
  }

  // override fun timeout(): Timeout = timeout

  async write(data: Uint8Array | string, byteCount?: number) {
    kotlinCheck(!this.closed, "closed")
    if (byteCount === 0) return
    let hexString = byteCount?.toString(16)
    this.sink.write(hexString, hexString?.length)
    this.sink.write("\r\n", 2)
    this.sink.write(data, byteCount)
    this.sink.write("\r\n", 2)
  }

  // @Synchronized
  async flush() {
    if (this.closed) return // Don't throw; this stream might have been closed on the caller's behalf.
    await this.sink.flush()
  }

  // @Synchronized
  async close() {
    if (this.closed) return
    this.closed = true
    this.sink.write("0\r\n\r\n", 5)
    // this.codec.detachTimeout(timeout)
    this.codec.state = Http1ExchangeCodec.STATE_READ_RESPONSE_HEADERS
  }

  buffer(): Sink {
    return this.sink
  }
}

export abstract class AbstractSource implements Source {
  protected closed: boolean = false;

  // 可在此处添加 timeout 支持（略）
  // protected timeout: Timeout = new ForwardingTimeout(source.timeout());

  constructor(protected http1Codec: Http1ExchangeCodec) {}

  Stream(callback: (chunk: Uint8Array, complete?:  boolean, error?: Error) => void) {
    throw new Error("Method not implemented.")
  }

  readString(): Promise<string> {
    throw new Error("Method not implemented.")
  }

  readBytes(): Promise<Uint8Array> {
    throw new Error("Method not implemented.")
  }

  _isGzip: boolean = false;
  set isGzip(value: boolean) {
    this._isGzip = value
  }

  /**
   * 读取数据，捕获异常并标记连接不可复用
   */
  async read(sink: Uint8Array, byteCount: number): Promise<number> {
    try {
      return this.http1Codec.source.read(sink, byteCount);
    } catch (e) {
      this.http1Codec.connection.noNewExchanges()
      this.responseBodyComplete();
      throw e;
    }
  }

  /**
   * 标记响应体读取完成
   * 用于释放连接、允许复用
   */
  responseBodyComplete(): void {
    if (this.http1Codec.state === Http1ExchangeCodec.STATE_CLOSED) return;
    if (this.http1Codec.state !== Http1ExchangeCodec.STATE_READING_RESPONSE_BODY) {
      throw new IllegalStateException(`state: ${this.http1Codec.state}`);
    }
    // this.http1Codec.detachTimeout(this.timeout)
    this.http1Codec.state = Http1ExchangeCodec.STATE_CLOSED;
  }


  skipAll(): void {
    this.close()
  }

  buffer(): BufferedSource {
    return this.http1Codec.source.buffer();
  }

  /**
   * 关闭源
   */
  close(): void {
    if (this.closed) return;
    this.closed = true;
    this.http1Codec.source.close()
  }
}

export class FixedLengthSource extends AbstractSource {
  private bytesRemaining: number;

  constructor(protected http1Codec: Http1ExchangeCodec, bytesRemaining: number) {
    super(http1Codec);
    this.bytesRemaining = bytesRemaining;

    // 如果长度为 0，立即完成
    if (this.bytesRemaining === 0) {
      this.responseBodyComplete();
    }
  }

  override async read(sink: Uint8Array, byteCount: number): Promise<number> {
    kotlinRequire(byteCount >= 0, "byteCount < 0: $byteCount")
    kotlinCheck(!this.closed, "closed")
    if (this.bytesRemaining === 0) return -1

    let temp_read = await super.read(sink, minOf(this.bytesRemaining, byteCount))
    if (temp_read === -1) {
      this.http1Codec.connection.noNewExchanges() // The server didn't supply the promised content length.
      let e = new ProtocolException("unexpected end of stream")
      this.responseBodyComplete()
      throw e
    }

    this.bytesRemaining -= temp_read
    if (this.bytesRemaining == 0) {
      this.responseBodyComplete()
    }
    return temp_read
  }

  override async close(): Promise<void> {
    if (this.closed) return

    if (this.bytesRemaining !== 0 &&
      !await this.discard(DISCARD_STREAM_TIMEOUT_MILLIS)) {
      this.http1Codec.connection.noNewExchanges() // Unread bytes remain on the stream.
      this.responseBodyComplete()
    }

    this.closed = true
  }

  skipAll(){
    super.skipAll()
  }

  /**
   * 丢弃剩余的响应体数据
   * 防止污染下一次请求（HTTP pipelining 安全）
   */
  private async discard(timeoutMs: number): Promise<boolean> {
    const junk = new Uint8Array(4096);
    const startTime = Date.now();

    while (this.bytesRemaining > 0 && Date.now() - startTime < timeoutMs) {
      const toRead = Math.min(this.bytesRemaining, junk.length);
      try {
        const read = await this.http1Codec.source.read(junk, toRead);
        if (read === -1) break;
        this.bytesRemaining -= read;
      } catch (e) {
        return false;
      }
    }

    return this.bytesRemaining === 0;
  }
}

/** An HTTP body with alternating chunk sizes and chunk bodies. */
export class ChunkedSource extends AbstractSource{
  private bytesRemainingInChunk = Http1ExchangeCodec.NO_CHUNK_YET
  private hasMoreChunks = true
  constructor(protected http1Codec: Http1ExchangeCodec, protected url: HttpUrl) {
    super(http1Codec);
  }

  override async read(sink: Uint8Array, byteCount: number): Promise<number> {
    kotlinRequire(byteCount >= 0, "byteCount < 0: $byteCount")
    kotlinCheck(!this.closed, "closed")
    if (!this.hasMoreChunks) return -1
    if (this.bytesRemainingInChunk === 0 || this.bytesRemainingInChunk === Http1ExchangeCodec.NO_CHUNK_YET){
      await this.readChunkSize()
      if (!this.hasMoreChunks) return -1
    }
    let read = await super.read(sink, minOf(byteCount, this.bytesRemainingInChunk))
    if (read === -1) {
      this.http1Codec.connection.noNewExchanges() // The server didn't supply the promised chunk length.
      let e = new ProtocolException("unexpected end of stream")
      this.responseBodyComplete()
      throw e
    }
    this.bytesRemainingInChunk -= read
    return read
  }

  private async readChunkSize(){
    // Read the suffix of the previous chunk.
    if (this.bytesRemainingInChunk !== Http1ExchangeCodec.NO_CHUNK_YET) {
      await this.http1Codec.source.readUtf8LineStrict()
    }
    try {
      this.bytesRemainingInChunk = await this.http1Codec.source.readHexadecimalUnsignedLong()
      let extensions = (await this.http1Codec.source.readUtf8LineStrict()).trim()
      if (this.bytesRemainingInChunk < 0 || extensions && extensions !== "" && !extensions.startsWith(";")) {
        throw new ProtocolException("expected chunk size and optional extensions" +
          " but was \"$bytesRemainingInChunk$extensions\"")
      }
    } catch (e) {
      throw new ProtocolException(e.message)
    }
    if (this.bytesRemainingInChunk === 0) {
      this.hasMoreChunks = false
      this.http1Codec._trailers = await this.http1Codec.headersReader.readHeaders()
      receiveHeaders(this.http1Codec.client!!.cookieJar, this.url, this.http1Codec._trailers)
      this.responseBodyComplete()
    }
  }

  override close(): void {
    if (this.closed) return;
    if (this.hasMoreChunks &&
    !this.discard(DISCARD_STREAM_TIMEOUT_MILLIS)){
      this.http1Codec.connection.noNewExchanges();
      this.responseBodyComplete();
    }
    this.closed = true
  }

  /**
   * 丢弃剩余的响应体数据
   * 防止污染下一次请求（HTTP pipelining 安全）
   */
  private async discard(timeoutMs: number): Promise<boolean> {
    this.http1Codec.source.skipAll()
    return true;
  }
}

/** An HTTP message body terminated by the end of the underlying stream. */
export class UnknownLengthSource extends AbstractSource {
  private inputExhausted: boolean = false
  constructor(protected http1Codec: Http1ExchangeCodec) {
    super(http1Codec);
  }

  override async read(sink: Uint8Array, byteCount: number): Promise<number> {
    kotlinRequire(byteCount >= 0, `byteCount < 0: ${byteCount}`)
    kotlinCheck(!this.closed, "closed")
    if (this.inputExhausted) return -1

    let read = await super.read(sink, byteCount)
    if (read === -1) {
      this.inputExhausted = true
      this.responseBodyComplete()
      return -1
    }
    return read
  }

  override close() {
    if (this.closed) return
    if (!this.inputExhausted) {
      this.responseBodyComplete()
    }
    this.closed = true
  }
}